package org.example.udf;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.annotation.InputGroup;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * 自定义聚合函数（多进一出）
 */
public class MyAggregateFunctionDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //分数、权重
        DataStreamSource<Tuple3<String, Integer, Integer>> scoreWeightDS = env.fromElements(
                Tuple3.of("zs", 80, 3),
                Tuple3.of("zs", 90, 4),
                Tuple3.of("zs", 95, 4),
                Tuple3.of("ls", 75, 3),
                Tuple3.of("ls", 85, 4),
                Tuple3.of("ls", 65, 4)
        );

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //指定字段
        //Table table = tableEnv.fromDataStream(sensorDS,$("id"));
        Table scoreWeightTable = tableEnv.fromDataStream(scoreWeightDS,
                $("f0").as("name"), $("f1").as("score"), $("f2").as("weight"));
        tableEnv.createTemporaryView("scores", scoreWeightTable);

        //注册函数
        tableEnv.createTemporarySystemFunction("WeightAvg", WeightAvgFunction.class);

        //重命名字段
        tableEnv.sqlQuery("select name,WeightAvg(score,weight) from scores group by name ")
                .execute()
                .print();

    }

    //泛型：返回值类型、累加器类型<加权总和、权重总和>
    public static class WeightAvgFunction extends AggregateFunction<Double, Tuple2<Integer, Integer>> {

        @Override
        public Double getValue(Tuple2<Integer, Integer> acc) {
            return acc.f0 * 1D / acc.f1;
        }

        @Override
        public Tuple2<Integer, Integer> createAccumulator() {
            return Tuple2.of(0, 0);
        }

        //固定的累加器的类型
        public void accumulate(Tuple2<Integer, Integer> acc, Integer score, Integer weight) {
            acc.f0 += score * weight;
            acc.f1 += weight;
        }
    }
}
